package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.*;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.*;
import java.util.List;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private ISeckillVoucherService seckillVoucherService;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedissonClient redissonClient;

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT=new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }



    //创建线程池,单线程
    private ExecutorService SECKILL_EXECUTORS=Executors.newSingleThreadExecutor();

    //VoucherOrderServiceImpl初始化时该线程就会启动

    @PostConstruct
    public void init(){
        log.info("seckill线程池启动了！");
        SECKILL_EXECUTORS.submit(new VoucherOrderHandler());
    }

    private class VoucherOrderHandler implements Runnable{
        @Override
        public void run() {
            while (true) {
                try {
                    String queueName = "stream.orders";
                    List<MapRecord<String, Object, Object>> msg = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().block(Duration.ofSeconds(2)).count(1),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed())
                    );
                    //消息队列为空则继续等待
                    if (msg==null||msg.isEmpty()) {
                        continue;
                    }
                    //解析消息
                    MapRecord<String, Object, Object> record = msg.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    //处理完成后需要确认消息
                    stringRedisTemplate.opsForStream().acknowledge(queueName,"g1",record.getId());
                } catch (Exception e) {

                    log.error("阻塞队列异常,处理pending_list");
                    log.error(e.getMessage());
                    handlePendingList();
                    throw new RuntimeException(e);
                }
            }
        }

        private void handlePendingList() {
            while (true) {
                try {
                    String queueName = "stream.orders";
                    //处理pending_list的消息，不需要加block时间，读取偏移量为队列中第一个消息
                    List<MapRecord<String, Object, Object>> msg = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create(queueName, ReadOffset.from("0"))
                    );
                    if (msg==null||msg.isEmpty()) {
                        //pending_list为空，代表消息处理完毕，直接跳出循环
                        break;
                    }
                    //解析消息
                    MapRecord<String, Object, Object> record = msg.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    //处理完成后需要确认消息
                    stringRedisTemplate.opsForStream().acknowledge(queueName,"g1",record.getId());
                } catch (Exception e) {
                    log.error("处理pending_list时异常！");
                    log.error(e.getMessage());
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    }
                }
            }
        }
    }
    /*
    //阻塞队列
    private static final BlockingQueue<VoucherOrder> SECKILL_BLOCKING_QUEUE=new ArrayBlockingQueue<>(1024*1024);
    private class VoucherOrderHandler implements Runnable{
        @Override
        public void run() {
            while (true) {
                try {
                    VoucherOrder voucherOrder = SECKILL_BLOCKING_QUEUE.take();
                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    log.error("阻塞队列异常");
                    log.error(e.getMessage());
                    throw new RuntimeException(e);
                }
            }
        }
    }*/

    private void handleVoucherOrder(VoucherOrder voucherOrder) throws InterruptedException {
        //userId需要通过voucherOrder获取，因为进程不是在主线程进行的，不能用UserHolder
        Long userId = voucherOrder.getUserId();
        RLock redisLock = redissonClient.getLock("lock:order:" + userId);   //使用redisson
        boolean isLock = redisLock.tryLock(1L,30L, TimeUnit.SECONDS);   //第一个参数为最大重试时间，第二个为锁的超时释放时间
        if (!isLock) {
            log.info("一个人只允许下一单！");
            return;
        }
        //这里的代理对象需要在主线程先获取
        try {
            currentProxy.createVoucherOrder(voucherOrder);
        } catch (IllegalStateException e) {
            throw new RuntimeException(e);
        } finally {
            //释放锁
            redisLock.unlock();
        }
    }

    /*
    @Override
    @Transactional(rollbackFor = Exception.class)
    public Result seckillVoucher(Long voucherId) {
        //查询优惠券信息
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //优惠券是否存在
        if (voucher == null) {
            return Result.fail("优惠券不存在！");
        }
        //判断秒杀是否开始
        LocalDateTime now = LocalDateTime.now();
        if (now.isBefore(voucher.getBeginTime())||now.isAfter(voucher.getEndTime())) {
            return Result.fail("活动未开始或已结束！");
        }
        //判断库存是否充足
        Integer stock = voucher.getStock();
        if (stock <= 0) {
            return Result.fail("库存不足！");
        }

        Long userId = UserHolder.getUser().getId();
        //1.给userId上锁，保证事务不会因为锁的释放而失效,先释放锁，再提交事务
        //  如果先释放锁再提交事务，会引起并发安全问题
        //  比如锁已被释放，事务还没提交，数据还没更新，但是另外一个线程获取锁并进行查询，数据未更新，会引起超卖问题

        //2.userId.toString().intern()表示jvm维护的字符串池中的userId值
        //举例：
        String str1 = new String("hello");
        String str2 = new String("hello");
        String str3 = "hello";
        String str4 = str1.intern();

        System.out.println(str1 == str2); // false - 不同对象
        System.out.println(str1 == str3); // false - 不同对象
        System.out.println(str3 == str4); // true - 都指向常量池中的同一个对象
        //相当于给字符串池中的这个值上锁
        synchronized (userId.toString().intern()){
            //为什么要使用代理对象？
            //因为如果直接调用，那就是这个VoucherOrderService对象本体调用方法
            //但是spring中的事务需要代理对象才能生效

            //为什么只有代理对象事务才能生效？
            // 1. Spring事务基于AOP代理实现，只有通过代理对象调用才会被事务拦截器处理
            // 2. 类内部方法调用（this.方法名()）会绕过代理，导致事务注解失效
            // 3. AopContext.currentProxy() 可以获取当前方法的代理对象
            IVoucherOrderService currentProxy =(IVoucherOrderService) AopContext.currentProxy();
            return currentProxy.createVoucherOrder(voucherId, now);
        }
    }
*/         //v1无法解决集群的下的线程安全问题

    //v2使用分布式锁
    /*
    @Override
    @Transactional(rollbackFor = Exception.class)
    public Result seckillVoucher(Long voucherId) throws InterruptedException {
        //查询优惠券信息
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //优惠券是否存在
        if (voucher == null) {
            return Result.fail("优惠券不存在！");
        }
        //判断秒杀是否开始
        LocalDateTime now = LocalDateTime.now();
        if (now.isBefore(voucher.getBeginTime())||now.isAfter(voucher.getEndTime())) {
            return Result.fail("活动未开始或已结束！");
        }
        //判断库存是否充足
        Integer stock = voucher.getStock();
        if (stock <= 0) {
            return Result.fail("库存不足！");
        }

        Long userId = UserHolder.getUser().getId();
        //创建锁对象
//        SimpleRedisLock redisLock = new SimpleRedisLock("order:"+userId,stringRedisTemplate);
        //尝试获取锁

        RLock redisLock = redissonClient.getLock("lock:order:" + userId);   //使用redisson
        boolean isLock = redisLock.tryLock(1L,30L, TimeUnit.SECONDS);   //第一个参数为最大重试时间，第二个为锁的超时释放时间
        if (!isLock) {
            return Result.fail("一个人只允许下一单！");
        }
        //为什么要使用代理对象？
        //因为如果直接调用，那就是这个VoucherOrderService对象本体调用方法
        //但是spring中的事务需要代理对象才能生效

        //为什么只有代理对象事务才能生效？
        // 1. Spring事务基于AOP代理实现，只有通过代理对象调用才会被事务拦截器处理
        // 2. 类内部方法调用（this.方法名()）会绕过代理，导致事务注解失效
        // 3. AopContext.currentProxy() 可以获取当前方法的代理对象
        try {
            IVoucherOrderService currentProxy = (IVoucherOrderService) AopContext.currentProxy();
            return currentProxy.createVoucherOrder(voucherId, now);
        } catch (IllegalStateException e) {
            throw new RuntimeException(e);
        } finally {

            //释放锁
            redisLock.unlock();
        }
    }
    */
    private IVoucherOrderService currentProxy;

    //v3使用lua脚本在redis中判断库存和一人一单增加性能，通过消息队列异步秒杀
    /*public Result seckillVoucher(Long voucherId) throws InterruptedException {
        Long userId = UserHolder.getUser().getId();
        //1.判断库存是否充足和一人一单问题
        Long res = stringRedisTemplate.execute(SECKILL_SCRIPT
                , Collections.emptyList()
                , voucherId.toString(), userId.toString());
        if (res == 1L){
            return Result.fail("库存不足！");
        }
        if (res == 2L){
            return Result.fail("一人只允许下一单！");
        }
        //到这里实际上购买资格的判断已经结束，把耗时高的保存订单过程放到阻塞队列中与资格判断异步进行
        //res为0，把下单信息保存到阻塞队列
        RedisIdWorker redisIdWorker = new RedisIdWorker(stringRedisTemplate);
        VoucherOrder voucherOrder = new VoucherOrder();
        Long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setUserId(userId);
        voucherOrder.setCreateTime(LocalDateTime.now());
        voucherOrder.setUpdateTime(LocalDateTime.now());
        currentProxy=(IVoucherOrderService) AopContext.currentProxy();
        SECKILL_BLOCKING_QUEUE.add(voucherOrder);
        return Result.ok(orderId);
    }*/

    //v4使用redis的stream类型来实现消息队列，可以实现消息的持久化，消息数量不会受jvm内存上限的影响
    public Result seckillVoucher(Long voucherId) throws InterruptedException {
        Long userId = UserHolder.getUser().getId();
        RedisIdWorker redisIdWorker = new RedisIdWorker(stringRedisTemplate);
        VoucherOrder voucherOrder = new VoucherOrder();
        Long orderId = redisIdWorker.nextId("order");
        //1.判断库存是否充足和一人一单问题
        Long res = stringRedisTemplate.execute(SECKILL_SCRIPT
                , Collections.emptyList()
                , voucherId.toString(), userId.toString(),orderId.toString());
        if (res == 1L){
            return Result.fail("库存不足！");
        }
        if (res == 2L){
            return Result.fail("一人只允许下一单！");
        }
        //到这里实际上购买资格的判断已经结束，把耗时高的保存订单过程放到消息队列中与资格判断异步进行提高效率
        //lua脚本中已经将消息添加到阻塞队列中
        currentProxy=(IVoucherOrderService) AopContext.currentProxy();
        return Result.ok(orderId);
    }
    @Transactional(rollbackFor = Exception.class)
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        //一人一单
        Long userId = voucherOrder.getUserId();
        Long voucherId = voucherOrder.getVoucherId();
        Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        if (count > 0) {
            log.error("一人只能买一单！");
            return;
        }
        //扣减库存
        boolean update = seckillVoucherService.update()
                .setSql("stock = stock - 1" )
                .eq("voucher_id", voucherId)
                .gt("stock", 0).update();
        if (!update) {
            log.error("库存不足！");
            return;
        }
        save(voucherOrder);
    }
}
